package com.sijibao.gather.producer.kafka;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KafkaProducerAdapter
{

	private static final Logger logger = LoggerFactory
			.getLogger(KafkaProducerAdapter.class);

	private KafkaProducerConfig kafkaProducerConfig;

	private KafkaProducer<String, String> producer;

	public void setKafkaProducerConfig(KafkaProducerConfig kafkaProducerConfig)
	{
		this.kafkaProducerConfig = kafkaProducerConfig;
	}

	private synchronized void init()
	{
		Map<String, Object> configs = new HashMap<String, Object>();
		configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
				kafkaProducerConfig.getBootstrapServers());
		configs.put(ProducerConfig.CLIENT_ID_CONFIG,
				kafkaProducerConfig.getClientId());
		producer = new KafkaProducer<String, String>(configs,
				new StringSerializer(), new StringSerializer());
	}

	public KafkaProducer<String, String> getProducer()
	{
		if (producer == null)
		{
			init();
		}
		return producer;
	}

	public KafkaProducerConfig getKafkaProducerConfig()
	{
		return this.kafkaProducerConfig;
	}

	public void sent(String value)
	{
		Future<RecordMetadata> future = getProducer()
				.send(new ProducerRecord<String, String>(
						getKafkaProducerConfig().getTopic(),
						UUID.randomUUID().toString(), value));
		try
		{
			future.get();
		} catch (Exception e)
		{
			logger.error(e.getMessage(), e);
		}
	}

	public void close()
	{
		if (producer != null)
		{
			producer.close();
			producer = null;
		}

	}

}
